package trouger.dmt;

import java.util.Iterator;

import org.apache.hadoop.mapred.JobConf;


class JoinableReducerEntity 
implements ReducerEntity<JoinableKey, JoinableValue>{

	//these indices are one-based.
	protected int[][] keyColumnList;
	protected int[][] valueColumnList;
	protected int[] maxColumnIndexOfSource;
	protected FieldIndexPair[] resultFieldsIndex;
	
	protected Class<? extends RecordGroupFilter> recordGroupFilterClass;
	protected AggregationOperation[] aggregationOperation;
	protected JoinType joinType;
	protected boolean firstSourceForceUnique;
	protected boolean secondSourceForceUnique;
	protected int maxReduceItems;
	
	public void configure(
			int[][] keyColumnList, 
			int[][] valueColumnList, 
			FieldIndexPair[] resultFieldsIndex,
			RecordGroupFilter recordGroupFilter,
			AggregationOperation[] aggregationOperation,
			JoinType joinType,
			boolean firstSourceForceUnique,
			boolean secondSourceForceUnique, 
			int maxItem){
		this.keyColumnList = keyColumnList;
		this.valueColumnList = valueColumnList;
		this.resultFieldsIndex = resultFieldsIndex;
		this.recordGroupFilterClass = recordGroupFilter.getClass();
		this.maxReduceItems = maxItem;

		int sourceCount = keyColumnList.length;
		maxColumnIndexOfSource = new int[sourceCount];
		for (int sourceIndex = 0; sourceIndex < sourceCount; sourceIndex++){
			int[] keyIndices = keyColumnList[sourceIndex];
			int[] valueIndices = valueColumnList[sourceIndex];
			int maxIndex = -1;
			for (int i = 0; i < keyIndices.length; i++){
				maxIndex = Math.max(maxIndex, keyIndices[i]);
			}
			for (int i = 0; i < valueIndices.length; i++){
				maxIndex = Math.max(maxIndex, valueIndices[i]);
			}
			maxColumnIndexOfSource[sourceIndex] = maxIndex;
		}
		
		this.aggregationOperation = aggregationOperation;
		this.joinType = joinType;
		this.firstSourceForceUnique = firstSourceForceUnique;
		this.secondSourceForceUnique = secondSourceForceUnique;
	}
	
	@Override
	public void acceptRecordGroup(JoinableKey key,
			Iterator<JoinableValue> values, RecordOutputer recordOutputer) {
		
		int sourceIndex = -1;	
		int[] keyIndices = null;
		int[] valueIndices = null;
		String[] record = null;
		RecordGroupFilter recordGroupFilter;
		
		try {
			recordGroupFilter = this.recordGroupFilterClass.newInstance();
		} catch (Exception e) {
			throw new IllegalStateException("cannot create record-group-filter");
		}
		if (recordGroupFilter.getClass().equals(JoinTableRecordGroupFilter.class)){
			JoinTableRecordGroupFilter filter = (JoinTableRecordGroupFilter)recordGroupFilter;
			filter.setResultFieldsIndex(resultFieldsIndex);
			filter.setAggregationOperations(aggregationOperation);
			if (joinType == null){
				joinType = JoinType.InnerJoin;
			}
			filter.setJoinType(joinType);
			filter.setForcedUniqueness(firstSourceForceUnique, secondSourceForceUnique);
		}else if (recordGroupFilter.getClass().equals(SelfAggregationRecordGroupFilter.class)){
			((SelfAggregationRecordGroupFilter)recordGroupFilter).setResultFieldsIndex(resultFieldsIndex);
			((SelfAggregationRecordGroupFilter)recordGroupFilter).setAggregationOperations(aggregationOperation);
		}
		recordGroupFilter.setRecordOutputer(recordOutputer);
		int count = 1;
		while (values.hasNext()){
			if (maxReduceItems!= 0 && count++ > maxReduceItems)
				return;
			JoinableValue v = values.next();
			if (sourceIndex != v.sourceIndex){
				keyIndices = keyColumnList[v.sourceIndex];
				valueIndices = valueColumnList[v.sourceIndex];
				record = new String[maxColumnIndexOfSource[v.sourceIndex] + 1];
				if (sourceIndex >= 0){
					if (!recordGroupFilter.endRecordSource()){
						return;
					}
				}
				if (!recordGroupFilter.beginRecordSource(v.sourceIndex + 1)){
					return;
				}
				sourceIndex = v.sourceIndex;
			}
			for (int i = 0; i < key.columnCount; i++){
				record[keyIndices[i]] = key.keyColumn[i];
			}
			for (int i = 0; i < v.columnCount; i++){
				record[valueIndices[i]] = v.valueColumn[i];
			}
			if (!recordGroupFilter.processRecord(record)){
				return;
			}
		}
		if (!recordGroupFilter.endRecordSource()){
			return;
		}
		recordGroupFilter.endOfGroup();
	}
}

class AggregationOperation{

	/**
	 * one-based index into the result fields to put the aggregated value
	 */
	private int resultField;
	/**
	 * one-based index into the source field to be aggregated
	 */
	private FieldIndexPair sourceField;
	
	private Class<? extends AggregationOperator> operatorClass;
	private boolean resumeOnError;
	/**
	 * one-based index into the result fields to put the aggregated value
	 */
	public int getResultField() {
		return resultField;
	}
	/**
	 * one-based index into the source field to be aggregated
	 */
	public FieldIndexPair getSourceField(){
		return sourceField;
	}
	public Class<? extends AggregationOperator> getOperatorClass(){
		return operatorClass;
	}
	public boolean isResumeOnError(){
		return resumeOnError;
	}
	
	public AggregationOperation(
			int resultField,
			FieldIndexPair sourceField, 
			Class<? extends AggregationOperator> operatorClass,
			boolean resumeOnError){
		this.resultField = resultField;
		this.sourceField = sourceField;
		this.operatorClass = operatorClass;
		this.resumeOnError = resumeOnError;
	}

}

class JoinableReducer 
extends BaseReducer<JoinableReducerEntity, JoinableKey, JoinableValue>{
	
	/*
	 * a list of strings, each in the format "className \t resumeOnError \t [sourceIndex.columnIndex]+".
	 * for example: "trouger.dmt.DefaultAggregationOperator    false    1.2 2.3".
	 * all indices are one based.
	 */
	public static String AGGREGATION_OPERATIONS_PARAM_NAME = "trouger.dmt.reducer.joinable.aggregation";
	/*
	 * "0" indicates to join tables, "1" indicates to do self aggregation
	 */
	public static String JOINTABLE_OR_SELFAGGREGATION_PARAM_NAME = "trouger.dmt.reducer.joinable.jointable-or-selfaggregation";
	
	protected enum AlterFunction{
		JoinTable,
		SelfAggregation
	}
	protected AlterFunction alterFunc;
	
	@SuppressWarnings("unchecked")
	@Override 
	public void configure(JobConf conf){
		super.configure(conf);
		
		//alter function
		String alterFuncStr = conf.get(JOINTABLE_OR_SELFAGGREGATION_PARAM_NAME);
		if (alterFuncStr == null){
			throw new IllegalArgumentException("join-table or self-aggregation? parameter missing");
		}
		if (alterFuncStr.equals("0")){
			alterFunc = AlterFunction.JoinTable;
		}else if (alterFuncStr.equals("1")){
			alterFunc = AlterFunction.SelfAggregation;
		}else{
			throw new IllegalArgumentException("invalid alter-function parameter: " + alterFuncStr);
		}

		//aggregation operations
		AggregationOperation[] aggregationOperation = null;
		String[] aggregationStrArray = BaseMapredUtils.getConfigStrings(conf, AGGREGATION_OPERATIONS_PARAM_NAME);
		if (aggregationStrArray != null){
			aggregationOperation = new AggregationOperation[aggregationStrArray.length];
			int k = 0;
			for (String s : aggregationStrArray){
				String[] part = s.split("\t", 3);
				try {
					Class<? extends AggregationOperator> operatorClass = 
						(Class<? extends AggregationOperator>) Class.forName(part[0]);
					boolean resumeOnError = Boolean.parseBoolean(part[1]);
					int resultField = Integer.parseInt(part[2]);
					aggregationOperation[k] = new AggregationOperation(
							resultField, 
							this.resultFieldsIndex[resultField - 1], 
							operatorClass, 
							resumeOnError);
					k++;
				} catch (ClassNotFoundException e) {
					throw new IllegalArgumentException("on loading AggregationOperator class, " + e.toString());
				}
			}
		}
		
		//default record group filter
		if (recordGroupFilter == null){
			if (alterFunc.equals(AlterFunction.JoinTable)){
				/*
				 * default join operation with more than two source tables is currently not supported
				 */
				if (sourceCount > 2){
					throw new UnsupportedOperationException("default join operation with more than two source tables is currently not supported");
				}else if (sourceCount < 2){
					throw new IllegalArgumentException("default join operation needs at least two source tables");
				}
				recordGroupFilter = new JoinTableRecordGroupFilter();
			}else if (alterFunc.equals(AlterFunction.SelfAggregation)){
				if (sourceCount != 1){
					throw new IllegalArgumentException("self-aggregation is specified but source-count is not one");
				}
				recordGroupFilter = new SelfAggregationRecordGroupFilter();
			}
		}
		
		//join type
		JoinType joinType = null;
		String joinTypeStr = conf.get(JoinTableRunner.JOINTYPE_PARAM_NAME);
		if (joinTypeStr != null){
			joinType = JoinType.valueOf(joinTypeStr);
		}
		
		//force unique
		String firstForceUniqueStr = conf.get(JoinTableRunner.FIRST_FORCE_UNIQUE_PARAM_NAME);
		String secondForceUniqueStr = conf.get(JoinTableRunner.SECOND_FORCE_UNIQUE_PARAM_NAME);
		boolean firstForceUnique;
		boolean secondForceUnique;
		if (firstForceUniqueStr != null){
			firstForceUnique = Boolean.parseBoolean(firstForceUniqueStr);
		}else{
			firstForceUnique = false;
		}
		if (secondForceUniqueStr != null){
			secondForceUnique = Boolean.parseBoolean(secondForceUniqueStr);
		}else{
			secondForceUnique = false;
		}
		
		this.reducerEntity.configure(
			keyColumnList,
			valueColumnList,
			resultFieldsIndex,
			recordGroupFilter,
			aggregationOperation,
			joinType,
			firstForceUnique,
			secondForceUnique,
			maxReduceItem);
	}
}
